﻿/*======================================================================
== Copyright : BlueCurve (c)
== Licence   : Gnu/GPL v2.x
== Author    : Teddy Albina
== Email     : bluecurveteam@gmail.com
== Web site  : http://www.codeplex.com/BlueCurve
========================================================================*/
using System;
using System.Collections;
using System.IO;
using System.Runtime.Serialization.Formatters.Binary;
using System.Threading;
using BlueCurve.Common.Threads;
using BlueCurve.IndexationContract;
using Indexer.Remoting;

namespace Indexer
{
    /// <summary>
    /// Class de traitement des donées
    /// </summary>
    public class DataTreatment
    {
        #region 'Configuration'

        /// <summary>
        /// s_Instance de la class
        /// </summary>
        private static volatile DataTreatment s_Instance;
        /// <summary>
        /// Objet de synchronisation
        /// </summary>
        private static readonly object s_ThreadsLock = new object();
        /// <summary>
        /// Pile contenant les données à envoyer aux indexeurs
        /// </summary>
        private Queue m_Queue = new Queue();
        /// <summary>
        /// Permet de laisse tourner le thread
        /// </summary>
        private bool m_IsRun = true;
        /// <summary>
        /// Nombre de thread actifs
        /// </summary>
        private long m_AliveThreads = 1;
        /// <summary>
        /// Pool de threads
        /// </summary>
        private SimpleLockThreadPool m_ThreadPool = new SimpleLockThreadPool(Environment.ProcessorCount * 15);

        /// <summary>
        /// Permet de stoper l'application
        /// </summary>
        public bool IsStop
        {
            set { this.m_IsRun = value; }
        }

        /// <summary>
        /// Retourne le nombre de threads actifs
        /// </summary>
        public long AliveThreads
        {
            get { return Interlocked.Read(ref this.m_AliveThreads); }
        }

        /// <summary>
        /// Retourne l'instance de la class
        /// </summary>
        public static DataTreatment GetInstance
        {
            get
            {
                if (s_Instance == null)
                {
                    lock (s_ThreadsLock)
                    {
                        if (s_Instance == null)
                        {
                            s_Instance = new DataTreatment();
                        }
                    }
                }
                return s_Instance;
            }
        }

        #endregion




        #region 'Functions'

        /// <summary>
        /// Lance le traitement des données de la pile
        /// </summary>
        public void RunTreatment()
        {
            this.Reload();
            while (this.m_IsRun)
            {
                lock (this.m_Queue.SyncRoot)
                {
                    while (this.m_Queue.Count > 0)
                    {
                        this.m_ThreadPool.QueueUserWorkItem(new WaitCallback(DoWork), this.m_Queue.Dequeue());
                    }
                }

                Thread.Sleep(20 * 1000);
            }
            while (this.AliveThreads > 1) { Thread.Sleep(150); }
            if (this.m_Queue.Count > 0)
                this.Save();

            Interlocked.Decrement(ref this.m_AliveThreads);
        }


        /// <summary>
        /// Fonction attachée au pool de threads
        /// </summary>
        /// <param name="data">Données de travail du thread</param>
        private void DoWork(object data)
        {
            Interlocked.Increment(ref this.m_AliveThreads);
            IndexationData document = (IndexationData)data;
            try
            {
                Tcpclient client = new Tcpclient();
                bool res = client.IndexDocument(document);

                if (!res)
                    throw new Exception();
            }
            catch
            {
                this.Add(document);
            }
            Interlocked.Decrement(ref this.m_AliveThreads);
        }


        /// <summary>
        /// Ajoute un élement dans la pile d'indexation
        /// </summary>
        /// <param name="data">Donnée à ajouter</param>
        public void Add(IndexationData data)
        {
            lock (this.m_Queue.SyncRoot)
            {
                this.m_Queue.Enqueue(data);
            }
        }


        /// <summary>
        /// Sauvegarde la pile d'indexation
        /// </summary>
        private void Save()
        {
            using (FileStream stream = File.Open("DataTreatmentthis.m_Queue.bin", FileMode.Create, FileAccess.Write, FileShare.None))
            {
                BinaryFormatter formatter = new BinaryFormatter();
                formatter.Serialize(stream, this.m_Queue);
                stream.Close();
            }
        }

        
        /// <summary>
        /// Recharge la pile d'indexation
        /// </summary>
        private void Reload()
        {
            if (File.Exists("DataTreatmentthis.m_Queue.bin"))
            {
                System.Collections.Queue queue;
                using (FileStream stream = File.Open("DataTreatmentthis.m_Queue.bin", FileMode.Open, FileAccess.Read, FileShare.None))
                {
                    BinaryFormatter formatter = new BinaryFormatter();
                    queue = (System.Collections.Queue)formatter.Deserialize(stream);
                    stream.Close();
                }

                while (queue.Count > 0)
                {
                    this.m_Queue.Enqueue((IndexationData)queue.Dequeue());
                }

                File.Delete("DataTreatmentthis.m_Queue.bin");
            }
        }

        #endregion
    }
}
